/**
 * @file market_data_consumer.h
 * @brief 市场数据消费者类的定义
 * 
 * @details 该文件定义了市场数据消费者类，负责接收和处理来自交易所的市场数据更新。
 * 市场数据消费者通过多播套接字接收两种类型的市场数据流：增量更新流和快照流。
 * 它处理这些数据流，检测和恢复丢包，并将处理后的市场数据更新传递给交易引擎。
 * 
 * @author 原作者
 * @date 2023
 */

#pragma once

#include <functional>
#include <map>

#include "common/thread_utils.h"
#include "common/lf_queue.h"
#include "common/macros.h"
#include "common/mcast_socket.h"

#include "exchange/market_data/market_update.h"

namespace Trading {
  /**
   * @brief 市场数据消费者类
   * 
   * @details 该类负责接收和处理来自交易所的市场数据更新。
   * 它管理两个多播套接字，一个用于接收增量市场数据更新，另一个用于接收市场数据快照。
   * 该类实现了丢包检测和恢复机制，当检测到增量流中的数据包丢失时，
   * 会自动切换到恢复模式，使用快照流进行同步。
   */
  class MarketDataConsumer {
  public:
    /**
     * @brief 构造函数
     * 
     * @details 初始化市场数据消费者，设置客户端 ID、市场数据更新队列和网络参数。
     * 在构造过程中，初始化并连接到增量数据流的多播套接字，但不会立即连接快照流。
     * 
     * @param client_id 客户端 ID，用于标识并生成日志文件名
     * @param market_updates 市场数据更新的无锁队列指针，用于将处理后的市场数据传递给交易引擎
     * @param iface 网络接口名称，用于多播通信
     * @param snapshot_ip 快照流的多播 IP 地址
     * @param snapshot_port 快照流的端口号
     * @param incremental_ip 增量流的多播 IP 地址
     * @param incremental_port 增量流的端口号
     */
    MarketDataConsumer(Common::ClientId client_id, Exchange::MEMarketUpdateLFQueue *market_updates, const std::string &iface,
                       const std::string &snapshot_ip, int snapshot_port,
                       const std::string &incremental_ip, int incremental_port);

    /**
     * @brief 析构函数
     * 
     * @details 停止市场数据消费者线程，然后等待 5 秒以确保线程安全终止。
     * 这个等待时间是为了确保所有进行中的操作都能够完成，避免资源泄漏或其他问题。
     */
    ~MarketDataConsumer() {
      stop();

      using namespace std::literals::chrono_literals;
      std::this_thread::sleep_for(5s);
    }

    /**
     * @brief 启动市场数据消费者线程
     * 
     * @details 设置运行标志并创建一个新线程来执行市场数据消费者的主循环。
     * 该方法使用 Common::createAndStartThread 函数创建线程，并传入 run 方法作为线程入口点。
     * 如果线程创建失败，将通过 ASSERT 宏触发断言。
     */
    auto start() {
      run_ = true;
      ASSERT(Common::createAndStartThread(-1, "Trading/MarketDataConsumer", [this]() { run(); }) != nullptr, "Failed to start MarketData thread.");
    }

    /**
     * @brief 停止市场数据消费者线程
     * 
     * @details 通过将运行标志设置为 false 来停止市场数据消费者线程。
     * 这将导致主循环退出，从而终止线程的执行。
     * 注意这个方法不会等待线程结束，只是发出停止信号。
     * 
     * @return void 无返回值
     */
    auto stop() -> void {
      run_ = false;
    }

    /// Deleted default, copy & move constructors and assignment-operators.
    MarketDataConsumer() = delete;

    MarketDataConsumer(const MarketDataConsumer &) = delete;

    MarketDataConsumer(const MarketDataConsumer &&) = delete;

    MarketDataConsumer &operator=(const MarketDataConsumer &) = delete;

    MarketDataConsumer &operator=(const MarketDataConsumer &&) = delete;

  private:
    /**
     * @brief 跟踪增量市场数据流上的下一个预期序列号
     * 
     * @details 用于检测数据包的丢失或间隔。当收到的序列号与预期的不一致时，
     * 将触发恢复过程。初始值为 1，表示预期第一个数据包的序列号为 1。
     */
    size_t next_exp_inc_seq_num_ = 1;

    /**
     * @brief 无锁队列，用于将解码后的市场数据更新推送给交易引擎
     * 
     * @details 该指针指向一个外部创建的无锁队列，用于将处理后的市场数据更新
     * 传递给交易引擎进行处理。这种无锁队列设计确保了在多线程环境中的高效数据传输。
     */
    Exchange::MEMarketUpdateLFQueue *incoming_md_updates_ = nullptr;

    /**
     * @brief 运行标志，控制主循环的执行
     * 
     * @details 当该标志为 true 时，主循环将继续运行；当设置为 false 时，主循环将退出。
     * 声明为 volatile 以确保在多线程环境中的可见性。
     */
    volatile bool run_ = false;

    /**
     * @brief 时间字符串缓冲区，用于日志记录
     */
    std::string time_str_;
    
    /**
     * @brief 日志记录器实例，用于记录市场数据消费者的日志
     */
    Logger logger_;

    /**
     * @brief 增量和快照市场数据流的多播订阅者套接字
     * 
     * @details 这两个套接字用于接收来自交易所的市场数据。
     * incremental_mcast_socket_ 用于接收增量市场数据更新，
     * snapshot_mcast_socket_ 用于接收市场数据快照。
     */
    Common::McastSocket incremental_mcast_socket_, snapshot_mcast_socket_;

    /**
     * @brief 跟踪当前是否正在进行恢复/同步过程
     * 
     * @details 该标志指示是否正在与快照市场数据流进行恢复/同步。
     * 这可能是因为刚刚启动或者检测到数据包丢失。
     * 当该标志为 true 时，将使用快照流进行恢复。
     */
    bool in_recovery_ = false;

    /**
     * @brief 快照多播流的网络信息
     * 
     * @details 存储快照多播流的网络接口、IP 地址和端口号。
     * 这些信息在需要进行恢复时用于连接快照流。
     */
    const std::string iface_, snapshot_ip_;
    const int snapshot_port_;

    /**
     * @brief 存储来自快照和增量通道的市场数据更新的容器
     * 
     * @details 这些容器按序列号递增的顺序存储市场数据更新。
     * 在恢复过程中，这些容器用于收集和排序市场数据更新，
     * 以确保数据的完整性和顺序性。
     */
    typedef std::map<size_t, Exchange::MEMarketUpdate> QueuedMarketUpdates;
    QueuedMarketUpdates snapshot_queued_msgs_, incremental_queued_msgs_;

  private:
    /**
     * @brief 该线程的主循环
     * 
     * @details 从多播套接字读取并处理消息。主要的处理逻辑在 recvCallback() 和
     * checkSnapshotSync() 方法中。该方法在单独的线程中执行，持续轮询多播套接字
     * 以检查是否有新的市场数据到达。
     * 
     * @return void 无返回值
     */
    auto run() noexcept -> void;

    /**
     * @brief 处理市场数据更新
     * 
     * @details 处理从多播套接字接收到的市场数据更新。消费者需要使用套接字参数
     * 来确定该数据是来自快照流还是增量流。该方法负责处理数据包，检测序列号间隔，
     * 并在需要时启动恢复过程。
     * 
     * @param socket 接收到数据的多播套接字指针，用于区分数据来源
     * @return void 无返回值
     */
    auto recvCallback(McastSocket *socket) noexcept -> void;

    /**
     * @brief 将消息排队到容器中
     * 
     * @details 将市场数据更新消息排入 *_queued_msgs_ 容器中。第一个参数指定该更新
     * 是来自快照流还是增量流。该方法将根据消息的来源将其存储在相应的容器中，
     * 并检查是否可以进行快照同步。
     * 
     * @param is_snapshot 指示更新是否来自快照流的布尔值
     * @param request 市场数据更新消息指针
     */
    auto queueMessage(bool is_snapshot, const Exchange::MDPMarketUpdate *request);

    /**
     * @brief 通过订阅快照多播流开始快照同步过程
     * 
     * @details 当检测到数据包丢失或系统刚启动时，调用此方法开始快照同步过程。
     * 该方法清除已排队的消息，初始化并连接到快照多播流。
     * 
     * @return void 无返回值
     */
    auto startSnapshotSync() -> void;

    /**
     * @brief 检查是否可以从排队的市场数据更新中进行恢复/同步
     * 
     * @details 检查是否可以从快照和增量市场数据流中排队的市场数据更新中进行恢复/同步。
     * 该方法分析排队的消息，确定是否有完整的快照和足够的增量更新来恢复到当前状态。
     * 
     * @return void 无返回值
     */
    auto checkSnapshotSync() -> void;
  };
}
